package net.opentsdb.core;

import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import com.sun.deploy.net.MessageHeader;
import jdk.nashorn.internal.ir.RuntimeNode;
import net.opentsdb.meta.TSMeta;
import net.opentsdb.tools.StartupPlugin;
import net.opentsdb.tsd.Query;
import net.opentsdb.tsd.StorageExceptionHandler;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.PluginLoader;
import net.opentsdb.utils.Threads;
import net.opentsdb.utils.TsdStrUtil;
import org.hbase.async.Bytes;
import org.hbase.async.HBaseClient;
import org.hbase.async.PutRequest;
import org.jboss.netty.util.HashedWheelTimer;
import org.jboss.netty.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Hujf
 * @title: TSDB
 * @date 2021-09-10 11:10
 * @description: 核心类 用于连接hbase 实现数据的查询和增加
 */
public class TSDB {
    private static final Logger LOG = LoggerFactory.getLogger(TSDB.class);

    static final byte[] FAMILY = {'t'};
    static final String METRICS_QUAL = "metrics";
    static final short METRICS_WIDTH = 3;
    static final String TAG_NAME_QUAL = "tagk";
    static final short TAG_NAME_WIDTH = 3;
    static final String TAG_VALUE_QUAL = "tagv";
    static final short TAG_VALUE_WIDTH = 3;

    private static final AtomicLong datapoints_added = new AtomicLong();

    byte[] table;

    private HBaseClient client;

    private Config config;

    private StartupPlugin startup = null;

    /**
     * Timer used for various tasks such as idle timeouts or query timeouts
     */
    private final HashedWheelTimer timer;

    UniqueId metrics;
    UniqueId tag_names;
    UniqueId tag_values;

    /**
     * Plugin for dealing with data points that can't be stored
     */
    private StorageExceptionHandler storage_exception_handler = null;

    //操作模式
    final OperationMode mode;
    private byte[] meta_table;
    private CompactionQueue compactionq;

    public TSDB(final Config config) {
        this(null, config);
    }

    public TSDB(HBaseClient client, Config config) {
        this.config = config;
        if (client == null) {
            org.hbase.async.Config async_config;
            if (!TsdStrUtil.isNullOrEmpty(config.configLocation())) {
                try {
                    async_config = new org.hbase.async.Config(config.configLocation());
                } catch (IOException e) {
                    throw new RuntimeException("Failed to read the config file: " +
                            config.configLocation(), e);
                }
            } else {
                async_config = new org.hbase.async.Config();
            }
            async_config.overrideConfig("hbase.zookeeper.znode.parent", config.getString("tsd.storage.hbase.zk_basedir"));
            async_config.overrideConfig("hbase.zookeeper.quorum", config.getString("tsd.storage.hbase.zk_quorum"));
            this.client = new HBaseClient(async_config);
        } else {
            this.client = client;
        }
        timer = Threads.newTimer("TSDB Timer");

        String string_mode = config.getString("tsd.mode");
        if (TsdStrUtil.isNullOrEmpty(string_mode)) {
            mode = OperationMode.READWRITE;
        } else if (string_mode.toLowerCase().equals("ro")) {
            mode = OperationMode.READONLY;
        } else if (string_mode.toLowerCase().equals("wo")) {
            mode = OperationMode.WRITEONLY;
        } else {
            mode = OperationMode.READWRITE;
        }

        table = config.getString("tsd.storage.hbase.data_table").getBytes(StandardCharsets.UTF_8);
        byte[] uidtable = config.getString("tsd.storage.hbase.uid_table").getBytes(StandardCharsets.UTF_8);

        meta_table = config.getString("tsd.storage.hbase.meta_table").getBytes(StandardCharsets.UTF_8);

        tag_names = new UniqueId(this.client, uidtable, TAG_NAME_QUAL, TAG_NAME_WIDTH, false);
        tag_values = new UniqueId(this.client, uidtable, TAG_VALUE_QUAL, TAG_VALUE_WIDTH, false);



        if (config.getBoolean("tsd.core.uid.random_metrics")) {
            metrics = new UniqueId(this.client, uidtable, METRICS_QUAL, METRICS_WIDTH, true);
        } else {
            metrics = new UniqueId(this.client, uidtable, METRICS_QUAL, METRICS_WIDTH, false);
        }
        compactionq = new CompactionQueue(this);

    }

    public static short tagk_width() {
        return TAG_NAME_WIDTH;
    }
    public static short tagv_width() {
        return TAG_VALUE_WIDTH;
    }

    public void initializePlugins(boolean init_rpcs) {
        String plugin_path = config.getString("tsd.core.plugin_path");
        loadPluginPath(plugin_path);


    }

    public static void loadPluginPath(final String plugin_path) {
        if (plugin_path != null && !plugin_path.isEmpty()) {
            try {
                PluginLoader.loadJARs(plugin_path);
            } catch (Exception e) {
                LOG.error("Error loading plugins from plugin path: " + plugin_path, e);
                throw new RuntimeException("Error loading plugins from plugin path: " +
                        plugin_path, e);
            }
        }
    }

    public void preFetchHBaseMeta() {
    }

    public Deferred<ArrayList<Object>> checkNecessaryTablesExist() {
        ArrayList<Deferred<Object>> checks = new ArrayList<Deferred<Object>>(2);
        checks.add(client.ensureTableExists(config.getString("tsd.storage.hbase.data_table")));
        checks.add(client.ensureTableExists(config.getString("tsd.storage.hbase.uid_table")));
        if (config.enable_tree_processing()) {
            checks.add(client.ensureTableExists(config.getString("tsd.storage.hbase.tree_table")));
        }
        if (config.enable_realtime_ts() || config.enable_realtime_uid() || config.enable_tsuid_incrementing()) {
            checks.add(client.ensureTableExists(config.getString("tsd.storage.hbase.meta_table")));
        }
        return Deferred.group(checks);
    }

    public Config getConfig() {
        return config;
    }

    public OperationMode getMode() {
        return mode;
    }

    public Timer getTimer() {
        return timer;
    }

    public StorageExceptionHandler getStorageExceptionHandler() {
        return storage_exception_handler;
    }

    public Deferred<Object> addPoint(String metric, long timestamp, long value, Map<String, String> tags) {
        final byte[] v;
        if (Byte.MIN_VALUE <= value && value <= Byte.MAX_VALUE) {
            v = new byte[]{(byte) value};
        } else if (Short.MIN_VALUE <= value && value <= Short.MAX_VALUE) {
            v = Bytes.fromShort((short) value);
        } else if (Integer.MIN_VALUE <= value && value <= Integer.MAX_VALUE) {
            v = Bytes.fromInt((int) value);
        } else {
            v = Bytes.fromLong(value);
        }

        final short flags = (short) (v.length - 1);  // Just the length.
        return addPointInternal(metric, timestamp, v, tags, flags);
    }

    private Deferred<Object> addPointInternal(String metric, long timestamp, byte[] value, Map<String, String> tags, short flags) {
        // 确认标签和时间
        checkTimestampAndTags(metric, timestamp, value, tags, flags);

        final byte[] row = IncomingDataPoints.rowKeyTemplate(this, metric, tags);

        final byte[] qualifier = Internal.buildQualifier(timestamp, flags);

        return storeIntoDB(metric, timestamp, value, tags, flags, row, qualifier);
    }

    private Deferred<Object> storeIntoDB(String metric, long timestamp, byte[] value, Map<String, String> tags, short flags, byte[] row, byte[] qualifier) {

        final long base_time ;

        if((timestamp&Const.SECOND_MASK)!=0){
            base_time = (timestamp/1000) - ((timestamp/1000)%Const.MAX_TIMESPAN);
        }else {
            base_time = (timestamp - (timestamp%Const.MAX_TIMESPAN));
        }


        final class WriteCallback implements Callback<Deferred<Object>, Boolean> {
            @Override
            public Deferred<Object> call(Boolean allow) throws Exception {
                if (!allow) {

                }
                Bytes.setInt(row,(int) base_time,metrics.width()+Const.SALT_WIDTH());
                Deferred<Object> result = null;

                scheduleForCompaction(row,(int) base_time);
                final PutRequest point = RequestBuilder.buildPutRequest(config, table, row, FAMILY, qualifier, value, timestamp);
                result = client.put(point);

                datapoints_added.incrementAndGet();

                final byte[] tsuid = UniqueId.getTSUIDFromKey(row,METRICS_WIDTH,Const.TIMESTAMP_BYTES);
                TSMeta.storeIfNecessary(TSDB.this,tsuid);

                return result;
            }
        }


        return Deferred.fromResult(true).addCallbackDeferring(new WriteCallback());
    }

    private void scheduleForCompaction(byte[] row, int base_time) {
        if(config.enable_compaction()){
            compactionq.add(row);
        }
    }

    private void checkTimestampAndTags(String metric, long timestamp, byte[] value, Map<String, String> tags, short flags) {
        if (timestamp < 0 || ((timestamp & Const.SECOND_MASK) != 0 &&
                timestamp > 9999999999999L)) {
            throw new IllegalArgumentException((timestamp < 0 ? "negative " : "bad")
                    + " timestamp=" + timestamp
                    + " when trying to add value=" + Arrays.toString(value) + '/' + flags
                    + " to metric=" + metric + ", tags=" + tags);
        }

        IncomingDataPoints.checkMetricAndTags(metric, tags);
    }


    public Deferred<Object> addPoint(String metric, long timestamp, float value, Map<String, String> tags) {
        if (Double.isNaN(value) || Double.isInfinite(value)) {
            throw new IllegalArgumentException("value is NaN or Infinite: " + value
                    + " for metric=" + metric
                    + " timestamp=" + timestamp);
        }
        final short flags = Const.FLAG_FLOAT | 0x7;  // A float stored on 8 bytes.
        return addPointInternal(metric, timestamp,
                Bytes.fromLong(Double.doubleToRawLongBits(value)),
                tags, flags);
    }

    public Deferred<Object> addPoint(String metric, long timestamp, double value, Map<String, String> tags) {
        if (Double.isNaN(value) || Double.isInfinite(value)) {
            throw new IllegalArgumentException("value is NaN or Infinite: " + value
                    + " for metric=" + metric
                    + " timestamp=" + timestamp);
        }
        final short flags = Const.FLAG_FLOAT | 0x7;  // A float stored on 8 bytes.
        return addPointInternal(metric, timestamp,
                Bytes.fromLong(Double.doubleToRawLongBits(value)),
                tags, flags);
    }

    public void collectStats() {
    }

    public byte[] metaTable() {
        return this.meta_table;
    }

    public HBaseClient getClient() {
        return this.client;
    }

    public Query newQuery() {
        return new TsdbQuery(this);
    }

    public enum OperationMode {
        READWRITE,
        READONLY,
        WRITEONLY;
    }

    public final void setStartupPlugin(final StartupPlugin plugin) {
        startup = plugin;
    }


    public Deferred<byte[]> getUIDAsync(final UniqueId.UniqueIdType type,String name){
        switch (type){
            case METRIC:
                return metrics.getIdAsync(name);
            case TAGK:
                return tag_names.getIdAsync(name);
            case TAGV:
                return tag_values.getIdAsync(name);
        }
        return null;
    }
}
